package ScalaSpark

import org.apache.spark.{SparkConf, SparkContext}

object ScalaActionOperation {

  def main(args: Array[String]): Unit = {
    //reduce()
    //collect()
    //count()
    //take()
    countByKey()
  }

  def reduce(): Unit = {
    val conf = new SparkConf().setAppName("reduce").setMaster("local")
    val sc = new SparkContext(conf)
    val numbersArray = Array(1, 2, 3, 4, 5, 6, 7, 8)
    val numberRDD = sc.parallelize(numbersArray, 1)
    val numbers = numberRDD.reduce(_ + _)
    println(numbers)
  }

  def collect(): Unit = {
    val conf = new SparkConf().setAppName("collect").setMaster("local")
    val sc = new SparkContext(conf)
    val numbersArray = Array(1, 2, 3, 4, 5, 6, 7, 8)
    val numberRDD = sc.parallelize(numbersArray, 1)
    val numbers = numberRDD.map(num => num * 2)
    val doubleNumberArray = numbers.collect()
    for (num <- doubleNumberArray) {
      println(num)
    }
  }

  def count(): Unit = {
    val conf = new SparkConf().setAppName("count").setMaster("local")
    val sc = new SparkContext(conf)
    val numbersArray = Array(1, 2, 3, 4, 5, 6, 7, 8)
    val numberRDD = sc.parallelize(numbersArray, 1)
    val count = numberRDD.count()
    println(count)
  }

  def take(): Unit = {
    val conf = new SparkConf().setAppName("take").setMaster("local")
    val sc = new SparkContext(conf)
    val numbersArray = Array(1, 2, 3, 4, 5, 6, 7, 8)
    val numberRDD = sc.parallelize(numbersArray, 1)
    val doubleNumberArray = numberRDD.take(3)
    for (num <- doubleNumberArray) {
      println(num)
    }
  }

  def countByKey(): Unit = {
    val conf = new SparkConf().setAppName("countByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val studentList = Array(new Tuple2[String, String]("class1", "aaa"),
      new Tuple2[String, String]("class2", "mack"),
      new Tuple2[String, String]("class1", "tom"),
      new Tuple2[String, String]("class2", "pos"))
    val scores = sc.parallelize(studentList, 1)
    val students = scores.countByKey()
    println(students)
  }
}
